package net.bwie.realtime.vehicle.dwd.function;

import net.bwie.realtime.utils.DateTimeUtil;
import net.bwie.realtime.vehicle.dwd.bean.AverageSpeed;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


/**
 * @author ASUS
 */
public class SpeedWindowFunction implements WindowFunction<MonitorInfo, AverageSpeed, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window,
                      Iterable<MonitorInfo> input, Collector<AverageSpeed> out) throws Exception {
        //存储总速度
        double sum = 0;
        //存储车数量
        int sum2 = 0;
        for (MonitorInfo monitorInfo : input) {
            sum += monitorInfo.getSpeed();
            sum2 += 1;
        }

        long start = window.getStart();
        long end = window.getEnd();
        String windowStartTime = DateTimeUtil.convertLongToString(start, "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(end, "yyyy-MM-dd HH:mm:ss");

        // 平均车速=当前窗口内通过车辆的车速之和 / 当前窗口内通过的车辆数量
        double avgSpeed = sum / sum2;

        out.collect(new AverageSpeed(windowStartTime, windowEndTime, s, sum / sum2, sum2));

    }
}
